package org.onion.pattern.event.springboot.rocketmq.consume.tt;//package org.ddd.pattern.event.spring.external.rocketmq.consume;
//
//import org.apache.rocketmq.client.consumer.store.OffsetStore;
//import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
//import org.apache.rocketmq.client.exception.MQBrokerException;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.client.impl.FindBrokerResult;
//import org.apache.rocketmq.client.impl.factory.MQClientInstance;
//import org.apache.rocketmq.common.MixAll;
//import org.apache.rocketmq.common.UtilAll;
//import org.apache.rocketmq.common.message.MessageQueue;
//import org.apache.rocketmq.remoting.exception.RemotingException;
//import org.apache.rocketmq.remoting.protocol.header.QueryConsumerOffsetRequestHeader;
//import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetRequestHeader;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import java.util.HashMap;
//import java.util.HashSet;
//import java.util.Map;
//import java.util.Set;
//import java.util.concurrent.ConcurrentHashMap;
//import java.util.concurrent.ConcurrentMap;
//import java.util.concurrent.atomic.AtomicLong;
//
///**
// * <p>
// * 自定义偏移量获取,CONSUME_FROM_LAST_OFFSET无法获取最新偏移量位置
// * 因为读取最新偏移量位置，ROCKETMQ本身机制为了保证消息尽可能被消费
// * 如果在消费者组中新起TOPIC进行订阅，会收到之前的消息进行消费，如果此时执行业务操作，容易发生数据问题
// */
//public class CustomRemoteBrokerOffsetStore implements OffsetStore {
//
//    private final static Logger log = LoggerFactory.getLogger(CustomRemoteBrokerOffsetStore.class);
//    private final MQClientInstance mQClientFactory;
//    private final String groupName;
//    private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
//            new ConcurrentHashMap<MessageQueue, AtomicLong>();
//
//
//    public CustomRemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName) {
//        this.mQClientFactory = mQClientFactory;
//        this.groupName = groupName;
////    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
//    }
//
//    @Override
//    public void load() {
//    }
//
//    @Override
//    public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) {
//        if (mq != null) {
//            AtomicLong offsetOld = this.offsetTable.get(mq);
//            if (null == offsetOld) {
//                offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset));
//            }
//
//            if (null != offsetOld) {
//                if (increaseOnly) {
//                    MixAll.compareAndIncreaseOnly(offsetOld, offset);
//                } else {
//                    offsetOld.set(offset);
//                }
//            }
//        }
//    }
//
//    @Override
//    public void updateAndFreezeOffset(MessageQueue messageQueue, long l) {
//
//    }
//
//    @Override
//    public long readOffset(final MessageQueue mq, final ReadOffsetType type) {
//        if (mq != null) {
//            switch (type) {
//                case MEMORY_FIRST_THEN_STORE:
//                case READ_FROM_MEMORY: {
//                    AtomicLong offset = this.offsetTable.get(mq);
//                    if (offset != null) {
//                        return offset.get();
//                    } else if (ReadOffsetType.READ_FROM_MEMORY == type) {
//                        return -1;
//                    }
//                }
//                case READ_FROM_STORE: {
//                    try {
//                        long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);
//                        if (brokerOffset == 0) {
//                            return -1;
//                        }
//                        AtomicLong offset = new AtomicLong(brokerOffset);
//                        this.updateOffset(mq, offset.get(), false);
//                        return brokerOffset;
//                    }
//                    // No offset in broker
//                    catch (MQBrokerException e) {
//                        return -1;
//                    }
//                    //Other exceptions
//                    catch (Exception e) {
//                        log.warn("【领域事件】fetchConsumeOffsetFromBroker exception, " + mq, e);
//                        return -2;
//                    }
//                }
//                default:
//                    break;
//            }
//        }
//
//        return -1;
//    }
//
//    @Override
//    public void persistAll(Set<MessageQueue> mqs) {
//        if (null == mqs || mqs.isEmpty()) {
//            return;
//        }
//
//        final HashSet<MessageQueue> unusedMQ = new HashSet<MessageQueue>();
//
//        for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
//            MessageQueue mq = entry.getKey();
//            AtomicLong offset = entry.getValue();
//            if (offset != null) {
//                if (mqs.contains(mq)) {
//                    try {
//                        this.updateConsumeOffsetToBroker(mq, offset.get());
//                        log.trace("【领域事件】[persistAll] Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}",
//                                this.groupName,
//                                this.mQClientFactory.getClientId(),
//                                mq,
//                                offset.get());
//                    } catch (Exception e) {
//                        log.error("【领域事件】updateConsumeOffsetToBroker exception, " + mq.toString(), e);
//                    }
//                } else {
//                    unusedMQ.add(mq);
//                }
//            }
//        }
//
//        if (!unusedMQ.isEmpty()) {
//            for (MessageQueue mq : unusedMQ) {
//                this.offsetTable.remove(mq);
//                log.trace("remove unused mq, {}, {}", mq, this.groupName);
//            }
//        }
//    }
//
//    @Override
//    public void persist(MessageQueue mq) {
//        AtomicLong offset = this.offsetTable.get(mq);
//        if (offset != null) {
//            try {
//                this.updateConsumeOffsetToBroker(mq, offset.get());
//                log.trace("【领域事件】[persist] Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}",
//                        this.groupName,
//                        this.mQClientFactory.getClientId(),
//                        mq,
//                        offset.get());
//            } catch (Exception e) {
//                log.error("【领域事件】updateConsumeOffsetToBroker exception, " + mq.toString(), e);
//            }
//        }
//    }
//
//    @Override
//    public void removeOffset(MessageQueue mq) {
//        if (mq != null) {
//            this.offsetTable.remove(mq);
//            log.trace("【领域事件】remove unnecessary messageQueue offset. group={}, mq={}, offsetTableSize={}",
//                    this.groupName, mq,
//                    offsetTable.size());
//        }
//    }
//
//    @Override
//    public Map<MessageQueue, Long> cloneOffsetTable(String topic) {
//        Map<MessageQueue, Long> cloneOffsetTable = new HashMap<MessageQueue, Long>();
//        for (Map.Entry<MessageQueue, AtomicLong> entry : this.offsetTable.entrySet()) {
//            MessageQueue mq = entry.getKey();
//            if (!UtilAll.isBlank(topic) && !topic.equals(mq.getTopic())) {
//                continue;
//            }
//            cloneOffsetTable.put(mq, entry.getValue().get());
//        }
//        return cloneOffsetTable;
//    }
//
//    /**
//     * Update the Consumer Offset in one way, once the Master is off, updated to Slave, here need to
//     * be optimized.
//     */
//    private void updateConsumeOffsetToBroker(MessageQueue mq, long offset) throws RemotingException,
//            MQBrokerException, InterruptedException, MQClientException {
//        updateConsumeOffsetToBroker(mq, offset, true);
//    }
//
//    /**
//     * Update the Consumer Offset synchronously, once the Master is off, updated to Slave, here need
//     * to be optimized.
//     */
//    @Override
//    public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway)
//            throws RemotingException,
//            MQBrokerException, InterruptedException, MQClientException {
//        FindBrokerResult findBrokerResult = this.mQClientFactory
//                .findBrokerAddressInAdmin(mq.getBrokerName());
//        if (null == findBrokerResult) {
//            this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
//            findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
//        }
//
//        if (findBrokerResult != null) {
//            UpdateConsumerOffsetRequestHeader requestHeader = new UpdateConsumerOffsetRequestHeader();
//            requestHeader.setTopic(mq.getTopic());
//            requestHeader.setConsumerGroup(this.groupName);
//            requestHeader.setQueueId(mq.getQueueId());
//            requestHeader.setCommitOffset(offset);
//
//            if (isOneway) {
//                this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffsetOneway(
//                        findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
//            } else {
//                this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffset(
//                        findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
//            }
//        } else {
//            throw new MQClientException("【领域事件】The broker[" + mq.getBrokerName() + "] not exist", null);
//        }
//    }
//
//    private long fetchConsumeOffsetFromBroker(MessageQueue mq)
//            throws RemotingException, MQBrokerException,
//            InterruptedException, MQClientException {
//        FindBrokerResult findBrokerResult = this.mQClientFactory
//                .findBrokerAddressInAdmin(mq.getBrokerName());
//        if (null == findBrokerResult) {
//
//            this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
//            findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
//        }
//
//        if (findBrokerResult != null) {
//            QueryConsumerOffsetRequestHeader requestHeader = new QueryConsumerOffsetRequestHeader();
//            requestHeader.setTopic(mq.getTopic());
//            requestHeader.setConsumerGroup(this.groupName);
//            requestHeader.setQueueId(mq.getQueueId());
//
//            return this.mQClientFactory.getMQClientAPIImpl().queryConsumerOffset(
//                    findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
//        } else {
//            throw new MQClientException("【领域事件】The broker[" + mq.getBrokerName() + "] not exist", null);
//        }
//    }
//
//
//}
